/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.infrastructure.queue.provider;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import sanbing.example.dylike.infrastructure.queue.QueueAdmin;
import sanbing.example.dylike.infrastructure.queue.QueueConsumer;
import sanbing.example.dylike.infrastructure.queue.QueueProducer;
import sanbing.example.dylike.infrastructure.queue.discovery.ServiceInfoProvider;
import sanbing.example.dylike.infrastructure.queue.memory.InMemoryQueueConsumer;
import sanbing.example.dylike.infrastructure.queue.memory.InMemoryQueueProducer;
import sanbing.example.dylike.infrastructure.queue.memory.InMemoryStorage;
import sanbing.example.dylike.infrastructure.queue.rpc.DefaultQueueRequestTemplate;
import sanbing.example.dylike.infrastructure.queue.rpc.QueueRequestTemplate;
import sanbing.example.dylike.infrastructure.queue.settings.QueueCoreSettings;
import sanbing.example.dylike.infrastructure.queue.settings.QueueRpcSettings;

@Slf4j
@Component
@ConditionalOnExpression("'${queue.type:null}'=='memory' || '${service.type:null}'=='none'")
public class InMemoryQueueFactory implements QueueFactory {


    private final InMemoryStorage storage;
    private final QueueRpcSettings rpcSettings;
    private final QueueCoreSettings coreSettings;
    private final ServiceInfoProvider serviceInfoProvider;

    public InMemoryQueueFactory(InMemoryStorage storage, QueueRpcSettings rpcSettings, QueueCoreSettings coreSettings, ServiceInfoProvider serviceInfoProvider) {
        this.storage = storage;
        this.rpcSettings = rpcSettings;
        this.coreSettings = coreSettings;
        this.serviceInfoProvider = serviceInfoProvider;
    }

    @Override
    public QueueProducer createToClickHouseMsgProducer() {
        return new InMemoryQueueProducer(storage, coreSettings.getClickhouseTopic());
    }

    @Override
    public QueueRequestTemplate createSeckillRpcRequestTemplate() {
        InMemoryQueueProducer producerTemplate =
                new InMemoryQueueProducer(storage, rpcSettings.getSeckillRequestsTopic());

        InMemoryQueueConsumer consumerTemplate =
                new InMemoryQueueConsumer(storage, rpcSettings.getSeckillResponsesTopic() + "." + serviceInfoProvider.getServiceId());

        DefaultQueueRequestTemplate.DefaultQueueRequestTemplateBuilder templateBuilder = DefaultQueueRequestTemplate.builder();

        templateBuilder.queueAdmin(new QueueAdmin() {
            @Override
            public void createTopicIfNotExists(String topic, String properties) {
            }

            @Override
            public void destroy() {
            }

        });

        templateBuilder.requestTemplate(producerTemplate);
        templateBuilder.responseTemplate(consumerTemplate);
        templateBuilder.maxPendingRequests(rpcSettings.getMaxPendingRequests());
        templateBuilder.maxRequestTimeout(rpcSettings.getMaxRequestsTimeout());
        templateBuilder.pollInterval(rpcSettings.getResponsePollInterval());
        return templateBuilder.build();
    }

    @Override
    public QueueConsumer createSeckillRpcRequestConsumer() {
        return new InMemoryQueueConsumer(storage, rpcSettings.getSeckillRequestsTopic());
    }

    @Override
    public QueueProducer createSeckillResponseProducer() {
        return new InMemoryQueueProducer(storage, rpcSettings.getSeckillResponsesTopic());
    }
}
